package org.jetlinks.pro.things.impl.connector.script;

import groovy.lang.Binding;
import groovy.lang.GroovyClassLoader;
import groovy.lang.MissingPropertyException;
import io.netty.util.Recycler;
import lombok.EqualsAndHashCode;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.codehaus.groovy.control.CompilerConfiguration;
import org.codehaus.groovy.runtime.InvokerHelper;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.event.ThingEventMessage;
import org.jetlinks.core.things.ThingId;
import org.jetlinks.core.things.ThingType;
import org.jetlinks.core.utils.RecyclerUtils;
import org.jetlinks.pro.gateway.DeviceMessageUtils;
import org.jetlinks.pro.streaming.Streaming;
import org.springframework.util.ClassUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.function.Function3;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.function.BiFunction;

@EqualsAndHashCode(of = "scriptDigest")
@Slf4j
public class ScriptThingMessageStreaming<T> implements Streaming<ThingMessage, T, T> {

    public final static CompilerConfiguration configuration = new CompilerConfiguration(CompilerConfiguration.DEFAULT);

    private static final Set<String> denyClass = new HashSet<>(
        Arrays.asList(System.class.getName(), Runtime.class.getName())
    );

    private static final GroovyClassLoader loader;

    static {
        configuration.setScriptBaseClass("org.jetlinks.pro.things.impl.connector.script.ThingMessageScript");
        loader = new GroovyClassLoader(new AccessControlClassLoader(ClassUtils.getDefaultClassLoader()), configuration);
    }

    @Setter
    private Class<ThingMessageScript> scriptType;

    private String scriptDigest;

    private BiFunction<ThingMessage, Object, T> converter;

    private Scheduler scheduler = Schedulers.boundedElastic();

    private final Recycler<ThingMessageScript> pool = RecyclerUtils.newRecycler(ThingMessageScript.class, handle -> {
        try {
            ThingMessageScript script = scriptType.newInstance();
            script.setScheduler(scheduler);
            script.setHandle(handle);
            return script;
        } catch (Exception e) {
            return (ThingMessageScript) InvokerHelper.createScript(scriptType, new Binding());
        }
    }, 1);

    @SuppressWarnings("all")
    public static <T> ScriptThingMessageStreaming<T> create(String script, BiFunction<ThingMessage, Object, T> converter) {

        ScriptThingMessageStreaming streaming = new ScriptThingMessageStreaming();
        streaming.converter = converter;
        if (script.contains("//immediate")) {
            streaming.scheduler = Schedulers.immediate();
        }
        String scriptDigest = DigestUtils.md5Hex(script);
        streaming.scriptDigest = scriptDigest;
        streaming.scriptType = loader.parseClass(script, "ThingComputeScript_" + DigestUtils.md5Hex(scriptDigest));

        return streaming;
    }

    private static class AccessControlClassLoader extends ClassLoader {
        public AccessControlClassLoader(ClassLoader parent) {
            super(parent);
        }

        @Override
        protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
            handleAccess(name);
            return super.loadClass(name, resolve);
        }

        @Override
        public Class<?> loadClass(String name) throws ClassNotFoundException {
            handleAccess(name);
            return super.loadClass(name);
        }

        @SneakyThrows
        private void handleAccess(String name) {
            if (denyClass.contains(name)) {
                for (StackTraceElement stackTraceElement : Thread.currentThread().getStackTrace()) {
                    if (stackTraceElement.getClassName().startsWith("ThingMessageScript")) {
                        throw new IllegalAccessException(name);
                    }
                }
            }
        }
    }

    @Override
    @SuppressWarnings("all")
    public Mono<T> compute(ThingMessage data) {
        ThingMessageScript instance = getScript();
        boolean needRelease = true;
        try {
            String type = data.getThingType();
            String id = data.getThingId();
            Binding binding = new Binding() {

                @Override
                public Object getVariable(String name) {

                    return getVariables().get(name);

                }

                @Override
                public Object getProperty(String property) {
                    Object var = getVariable(property);
                    if (var != null) {
                        return var;
                    }
                    try {
                        return super.getProperty(property);
                    } catch (MissingPropertyException e) {
                        return null;
                    }
                }
            };
            DeviceMessageUtils
                .tryGetProperties(data)
                .ifPresent(props -> props.forEach(binding::setVariable));

            if (data instanceof ThingEventMessage) {
                Object eventData = ((ThingEventMessage) data).getData();
                if (eventData instanceof Map) {
                    ((Map<String, ?>) eventData).forEach(binding::setVariable);
                } else {
                    binding.setVariable("data", eventData);
                }
            }

            binding.setVariable("message", data);
            binding.setVariable("thingType", type);
            binding.setVariable("thingId", id);
            //兼容device
            binding.setVariable("deviceId", id);
            binding.setVariable("timestamp", data.getTimestamp());
            binding.setVariable("__timestamp__", data.getTimestamp());
            instance.setBinding(binding);
            instance.setMessage(data);
            instance.setDefaultThingType(ThingType.of(type));
            return instance
                .runAsync()
                .onErrorResume(err -> {
                    log.error("execute script error:{}", err.getMessage(), err);
                    return Mono.empty();
                })
                .doFinally(s -> instance.release())
                .map(obj -> converter.apply(data, obj));
        } catch (Throwable e) {
            instance.release();
            return Mono.error(e);
        }
    }

    private ThingMessageScript getScript() {
        try {
            return pool.get();
        } catch (Throwable e) {
            return (ThingMessageScript) InvokerHelper.createScript(scriptType, new Binding());
        }
    }

    @Override
    public Flux<T> output() {
        return Flux.empty();
    }

    @Override
    public void dispose() {

    }

}
